package dim;

import java.io.File;
import java.util.Properties;
import java.util.Arrays;
import java.time.Duration;
import java.util.Map;
import java.util.HashMap;
import java.util.List;
import java.util.ArrayList;

import java.net.InetSocketAddress;

import java.util.regex.Pattern;
import java.util.regex.Matcher;

import java.net.NetworkInterface;
import java.net.InetAddress;
import java.util.Collections;

import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericData;

import io.github.cdimascio.dotenv.Dotenv;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.CqlSessionBuilder;
import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.cql.Row;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;

public class App {
    private static final String MODULE = "push-gate-java:batch";
    private static final Logger LOGGER = LogManager.getLogger(LogManager.ROOT_LOGGER_NAME);
    private static final ObjectMapper objectMapper = new ObjectMapper();

    private static void info(String message) {
        LOGGER.info("[component:{}] {}", MODULE, message);
    }

    private static void warn(String message) {
        LOGGER.warn("[component:{}] {}", MODULE, message);
    }

    private static void error(String message) {
        LOGGER.error("[component:{}] {}", MODULE, message);
    }

    private static void error(String message, Throwable e) {
        LOGGER.error("[component:" + MODULE + "] " + message, e);
    }

    private static boolean isInvalid(String value) {
        return value == null || "[\"\"]".equals(value) || "[\"null\"]".equals(value) || "[\"(null)\"]".equals(value) || "[\"null\",\"null\"]".equals(value);
    }

    private static String _resolveEnv(Dotenv dotenv, String name) {
        String v = dotenv.get(name);
        if (v == null)
            return null;
        Matcher p = Pattern.compile("\\$\\{([^}]*)\\}").matcher(v);

        String ret = v;
        while (p.find()) {
            String subVarValue = _resolveEnv(dotenv, p.group(1));
            if (subVarValue == null)
                return null;
            ret = v.replace(p.group(0), subVarValue);
        }
        return ret;
    }

    public static String resolveEnv(Dotenv dotenv, String name, String defaultValue) {
        String resolveValue = _resolveEnv(dotenv, name);
        return resolveValue != null ? resolveValue : defaultValue;
    }

    public static String resolveMyIp(String nodes) throws Exception {
        for (NetworkInterface ni : Collections.list(NetworkInterface.getNetworkInterfaces())) {
            for (InetAddress addr : Collections.list(ni.getInetAddresses())) {
                String currentIp = addr.getHostAddress();
                if (nodes.contains(currentIp)) {
                    return currentIp;
                }
            }
        }
        return null;
    }

    static class EnumState {
        public Integer maxEnumKey = null;
        public Map<Object, Integer> enumInfo = null;

        public EnumState() {
            this.maxEnumKey = 0;
            this.enumInfo = new HashMap<Object, Integer>();
        }
    }

    static class ClusterEnumState {
        public Map<Object, EnumState> clusterEnumState = null;

        public ClusterEnumState() {
            this.clusterEnumState = new HashMap<Object, EnumState>();
        }

        public EnumState get(Object key) {
            return this.clusterEnumState.get(key);
        }

        public void put(Object key, EnumState val) {
            this.clusterEnumState.put(key, val);
        }

        public void printMaxEnumKey() {
            for (Map.Entry<Object, EnumState> e : clusterEnumState.entrySet()) {
                info(String.format("key: %s -> %s", e.getKey(), e.getValue().maxEnumKey));
            }
        }
    }

    public static ClusterEnumState loadEnumState(CqlSession cqlSession, String tableName, String clusterSelectStr, String primarySelectStr, String enumValName) {
        String[] clusterSelect = clusterSelectStr == null ? new String[0] : clusterSelectStr.split(",");
        String[] primarySelect = clusterSelectStr == null ? new String[0] : primarySelectStr.split(",");

        ClusterEnumState clusterState = new ClusterEnumState();

        for (Row row : cqlSession.execute("select * from " + tableName)) {
            List<Object> clusterVals = new ArrayList<Object>();
            for (String clusterField : clusterSelect) {
                clusterVals.add(row.getObject(clusterField));
            }
            if (clusterState.get(clusterVals) == null) {
                clusterState.put(clusterVals, new EnumState());
            }
            EnumState enumState = clusterState.get(clusterVals);

            List<Object> enumKeys = new ArrayList<Object>();
            for (String primaryField : primarySelect) {
                enumKeys.add(row.getObject(primaryField));
            }
            int enumVal = row.getInt(enumValName);
            enumState.enumInfo.put(enumKeys, enumVal);
            if (enumState.maxEnumKey < enumVal) {
                enumState.maxEnumKey = enumVal;
            }
        }
        return clusterState;
    }

    public static String extractPlatformName(String schemaName) {
        String result = "unknown";
        if (schemaName.equals("DcSdkAndroid")) {
            result = "android";
        } else if (schemaName.equals("DcSdkIos")) {
            result = "ios";
        } else if (schemaName.equals("DcSdkWeb")) {
            result = "web";
        } else if (schemaName.equals("DcSdkWechat")) {
            result = "wechat";
        } else if (schemaName.equals("DcSdkH5")) {
            result = "H5";
        }
        return result;
    }

    public static void doEventEnumFromCommon(GenericRecord platform_record, Map<String, Object> collector) {
        String platformName = extractPlatformName(platform_record.getSchema().getName());
        GenericRecord common_record = (GenericRecord) platform_record.get("common");

        GenericRecord basic_info = (GenericRecord) common_record.get("basic");
        String project_id = String.valueOf(basic_info.get("project_id"));
        String app_id = String.valueOf(basic_info.get("app_id"));
        String tenant_id = String.valueOf(basic_info.get("tenant_id"));
        String app_version = String.valueOf(basic_info.get("app_version"));

        GenericRecord sdk_info = (GenericRecord) common_record.get("sdk_info");
        String sdk_version = null;
        if (sdk_info != null) {
            sdk_version = String.valueOf(sdk_info.get("sdk_version"));
        }
        collector.put("platform", platformName);
        collector.put("project_id", project_id);
        collector.put("app_id", app_id);
        collector.put("tenant_id", tenant_id);
        collector.put("app_version", app_version);
        collector.put("sdk_version", sdk_version);
    }

    public static Object avroGet(GenericRecord record, String path) {
        String[] parts = path.split("\\.");

        Object result = record;
        for (int i = 0; i < parts.length; i++) {
            result = ((GenericRecord) result).get(parts[i]);
            if (result == null) return null;
        }
        return result;
    }

    public static List<Map<String, Object>> doOptionEnumFromCommon(ClusterEnumState clusterEnumState, GenericRecord platform_record) throws Exception {
        List<Map<String, Object>> rets = new ArrayList<Map<String, Object>>();

//        String selectsStr = "basic.platform,basic.app_package_id,basic.app_version,basic.market_name,device_info.device_type,device_info.device_brand,device_info.device_model,os.os_system,carrier,browser.browser_type,screen.screen_height:screen.screen_width";
        String selectsStr = "sdk_info.sdk_version,sdk_info.sdk_type,device_info.user_agent,os.os_version,browser.browser_version," +
                "basic.platform,basic.app_package_id,basic.app_version,basic.market_name,device_info.device_type,device_info.device_brand,device_info.device_model,os.os_system,carrier,browser.browser_type,screen.screen_height:screen.screen_width";
        List<String> selects = Arrays.asList(selectsStr.split(","));

        String platform = extractPlatformName(platform_record.getSchema().getName());

        GenericRecord common_record = (GenericRecord) platform_record.get("common");
        GenericRecord basic_info = (GenericRecord) common_record.get("basic");

        String tenant_id = String.valueOf(basic_info.get("tenant_id"));

        for (String option_select : selects) {
            String cluster_select = "tenant_id";
            String cluster_val = new ObjectMapper().writeValueAsString(Arrays.asList(tenant_id));
            List<Object> selectVals = new ArrayList<Object>();
            for (String f : option_select.split(":")) {
                if (f.equals("platform")) {
                    selectVals.add(platform);
                } else {
                    selectVals.add(String.valueOf(avroGet(common_record, f)));
                }
            }

            String option_val = new ObjectMapper().writeValueAsString(selectVals);

            List<Object> clusterIndex = Arrays.asList(cluster_select, cluster_val);
            List<Object> optionIndex = Arrays.asList(option_select, option_val);

            if (clusterEnumState.get(clusterIndex) == null) {
                clusterEnumState.put(clusterIndex, new EnumState());
            }
            EnumState enumState = clusterEnumState.get(clusterIndex);

            if (enumState.enumInfo.get(optionIndex) != null) continue;
            if (isInvalid(option_val)) continue;
            enumState.maxEnumKey++;
            enumState.enumInfo.put(optionIndex, enumState.maxEnumKey);

            Map<String, Object> out_json = new HashMap<String, Object>();
            out_json.put("cluster_name", cluster_select);
            out_json.put("cluster_val", cluster_val);
            out_json.put("option_name", option_select);
            out_json.put("option_val", option_val);
            out_json.put("enum_key", enumState.maxEnumKey);
            rets.add(out_json);
        }
        return rets;
    }

    public static List<Map<String, Object>> doOptionEnumFromEvent(ClusterEnumState clusterEnumState, GenericRecord platform_record) throws Exception {
        List<Map<String, Object>> rets = new ArrayList<Map<String, Object>>();
        String selectsStr = "basic.event_code,basic.event_name,basic.event_type,basic.send_type,basic.page_type,basic.network_type,page_info.current_page_id,page_info.current_page_name,page_info.current_page_url,page_info.current_page_host,page_info.current_page_path,page_info.source_page_id,page_info.source_page_name,page_info.source_page_url,page_info.source_page_host,page_info.source_page_path," +
                "extensions.couponName,extensions.userrecordID,extensions.mallID,extensions.templateContentTitle,extensions.courseTitle," +
                "extensions.materialTitle,extensions.templateContentID,extensions.contentMaterialID,extensions.commonTestID,extensions.courseID," +
                "extensions.couponGroups,extensions.messageTitle,extensions.messageID,extensions.wechatAppID,extensions.responseLink,extensions.userSource," +
                "extensions.wechatHomePageID,extensions.activityID,extensions.templateID,extensions.couponType,extensions.couponCode,extensions.couponID," +
                "extensions.personnelType,extensions.associatedShopID,extensions.memberPlatform,extensions.integralMallID";
        List<String> selects = Arrays.asList(selectsStr.split(","));
        GenericRecord common_record = (GenericRecord) platform_record.get("common");
        GenericRecord common_basic_info = (GenericRecord) common_record.get("basic");
        String tenant_id = String.valueOf(common_basic_info.get("tenant_id"));
        for (GenericRecord event_record : (GenericData.Array<GenericRecord>) platform_record.get("events")) {
            OUTPUT:
            for (String option_select : selects) {
                String cluster_select = "tenant_id";
                String cluster_val = new ObjectMapper().writeValueAsString(Arrays.asList(tenant_id));
                List<Object> selectVals = new ArrayList<Object>();
                GenericRecord business_info = (GenericRecord) event_record.get("business_info");
                if ("extensions".equals(option_select.split("\\.")[0])) {
                    if (business_info != null) {
                        for (Map.Entry<Object, Object> e : ((Map<Object, Object>) business_info.get("extensions")).entrySet()) {
                            info(String.format("key: %s -> %s", e.getKey(), e.getValue()));
                            if (String.valueOf(e.getKey()).equals(option_select.split("\\.")[option_select.split("\\.").length - 1]))
                                if ("null".equals(String.valueOf(e.getValue()))) {
                                    continue OUTPUT;
                                } else {
                                    selectVals.add(String.valueOf(String.valueOf(e.getValue())));
                                }
                        }
                    }
                } else {
                    selectVals.add(String.valueOf(avroGet(event_record, option_select)));
                }
                String option_val = new ObjectMapper().writeValueAsString(selectVals);

                List<Object> clusterIndex = Arrays.asList(cluster_select, cluster_val);
                List<Object> optionIndex = Arrays.asList(option_select, option_val);

                if (clusterEnumState.get(clusterIndex) == null) {
                    clusterEnumState.put(clusterIndex, new EnumState());
                }
                EnumState enumState = clusterEnumState.get(clusterIndex);

                if (enumState.enumInfo.get(optionIndex) != null) continue;
                if (isInvalid(option_val)) continue;
                enumState.maxEnumKey++;
                enumState.enumInfo.put(optionIndex, enumState.maxEnumKey);

                Map<String, Object> out_json = new HashMap<String, Object>();
                out_json.put("cluster_name", cluster_select);
                out_json.put("cluster_val", cluster_val);
                out_json.put("option_name", option_select);
                out_json.put("option_val", option_val);
                out_json.put("enum_key", enumState.maxEnumKey);
                rets.add(out_json);
            }
        }
        return rets;
    }

    public static Map<String, Object> doEventEnumFromEvent(ClusterEnumState clusterState, GenericRecord event_record, Map<String, Object> collector) {
        if (clusterState.get(Arrays.asList()) == null) clusterState.put(Arrays.asList(), new EnumState());
        EnumState enumState = clusterState.get(Arrays.asList());


        GenericRecord basic_info = (GenericRecord) event_record.get("basic");
        String page_type = String.valueOf(basic_info.get("page_type"));
        String event_type = String.valueOf(basic_info.get("event_type"));
        String event_code = String.valueOf(basic_info.get("event_code"));
        // String event_code_md5 = event_record.hasField("event_code_md5") ? String.valueOf(event_record.get("event_code_md5")) : "";
        String event_code_md5 = String.valueOf(basic_info.get("event_code_md5"));
        String event_name = String.valueOf(basic_info.get("event_name"));
        String send_type = String.valueOf(basic_info.get("send_type"));


        String element_path = null, element_content = null;
        // if (event_record.hasField("element_info")) {
        GenericRecord element_info = (GenericRecord) event_record.get("element_info");
        if (element_info != null) {
            element_path = String.valueOf(element_info.get("element_path"));
            element_content = String.valueOf(element_info.get("element_content"));
        }
        //}


        String current_page_url = null, current_page_path = null, current_page_name = null;
        //if (event_record.hasField("page_info")) {
        GenericRecord page_info = (GenericRecord) event_record.get("page_info");
        if (page_info != null) {
            current_page_url = String.valueOf(page_info.get("current_page_url"));
            current_page_path = String.valueOf(page_info.get("current_page_path"));
            current_page_name = String.valueOf(page_info.get("current_page_name"));
        }
        //}

        GenericRecord business_info = (GenericRecord) event_record.get("business_info");

        Map<String, String> extensions = new HashMap<String, String>();
        if (business_info != null) {
            for (Map.Entry<Object, Object> e : ((Map<Object, Object>) business_info.get("extensions")).entrySet()) {
                extensions.put(String.valueOf(e.getKey()), String.valueOf(e.getValue()));
            }
        }

        String project_id = String.valueOf(collector.get("project_id"));
        String app_id = String.valueOf(collector.get("app_id"));
        String app_version = String.valueOf(collector.get("app_version"));
        List<Object> rowKey = Arrays.asList(project_id, app_id, app_version, event_code);

        if (enumState.enumInfo.get(rowKey) != null) return null;

        enumState.maxEnumKey++;
        enumState.enumInfo.put(rowKey, enumState.maxEnumKey);

        Map<String, Object> out_json = new HashMap<String, Object>();
        out_json.putAll(collector);
        out_json.put("event_enum_key", enumState.maxEnumKey);

        out_json.put("page_type", page_type);
        out_json.put("event_type", event_type);

        out_json.put("event_code", event_code);
        out_json.put("event_code_md5", event_code_md5);
        out_json.put("event_name", event_name);
        out_json.put("element_path", element_path);
        out_json.put("element_content", element_content);

        out_json.put("send_type", send_type);
        out_json.put("current_page_url", current_page_url);
        out_json.put("current_page_path", current_page_path);
        out_json.put("current_page_name", current_page_name);

        out_json.put("extensions", extensions);

        return out_json;
    }

    public static void writeEventEnum(Map<String, Object> out_json, Schema my_schema, KafkaProducer<byte[], GenericRecord> producer, String outTopic, CqlSession cqlSession) throws Exception {
        GenericRecord out_record = new GenericData.Record(my_schema);

        for (Map.Entry<String, Object> out_entry : out_json.entrySet()) {
            out_record.put(out_entry.getKey(), out_entry.getValue());
        }
        //producer.send(new ProducerRecord<byte[], GenericRecord>(outTopic, null, out_record)) ;

        producer.send(new ProducerRecord<byte[], GenericRecord>(outTopic, null, out_record), new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e != null) {
                    error("producer send fail ,topic:" + outTopic + " out_json : " + out_record + " Exception:", e);
                }
            }
        });

        String cql = "UPDATE event_enum_dim\n SET\n";
        try {
            boolean isFirst = true;
            for (Map.Entry<String, Object> out_entry : out_json.entrySet()) {
                if (out_entry.getKey().equals("event_enum_key")) continue;
                Object val = out_entry.getValue();
                String valStr = null;
                if (val instanceof String) {
                    valStr = "\'" + val.toString().replace("'", "\"") + "\'";
                } else {
                    //valStr = new ObjectMapper().writeValueAsString(val).replace("\"", "'") ;
                    valStr = objectMapper.writeValueAsString(val).replace("\"", "'");
                }

                if (isFirst) {
                    cql += String.format("    %s = %s\n", out_entry.getKey(), valStr);
                } else {
                    cql += String.format(",   %s = %s\n", out_entry.getKey(), valStr);
                }
                isFirst = false;
            }
            cql += "WHERE event_enum_key = " + out_json.get("event_enum_key");

            cqlSession.execute(cql);
        } catch (Exception e) {
            error(cql, e);
            e.printStackTrace();
        }
    }

    public static String formatVal(Object val) throws Exception {
        return (val instanceof String) ? "\'" + val.toString().replace("'", "\"") + "\'" : new ObjectMapper().writeValueAsString(val).replace("\"", "'");
    }

    public static void writeOptionEnum(Map<String, Object> out_json, Schema my_schema, KafkaProducer<byte[], GenericRecord> producer, String outTopic, CqlSession cqlSession) throws Exception {
        GenericRecord out_record = new GenericData.Record(my_schema);

        for (Map.Entry<String, Object> out_entry : out_json.entrySet()) {
            out_record.put(out_entry.getKey(), out_entry.getValue());
        }
        //producer.send(new ProducerRecord<byte[], GenericRecord>(outTopic, null, out_record)) ;
        producer.send(new ProducerRecord<byte[], GenericRecord>(outTopic, null, out_record), new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e != null) {
                    error("producer send fail ,topic:" + outTopic + " out_json : " + out_record + " Exception:", e);
                }
            }
        });

        String cql = "UPDATE option_enum\n SET\n"
                + String.format("    option_name = %s\n", formatVal(out_json.get("option_name")))
                + String.format(",   option_val = %s\n", formatVal(out_json.get("option_val")))
                + String.format("WHERE enum_key = %s\n", out_json.get("enum_key"))
                + String.format("  AND cluster_name = %s\n", formatVal(out_json.get("cluster_name")))
                + String.format("  AND cluster_val = %s", formatVal(out_json.get("cluster_val")));

        // info("cql:" + cql) ;
        cqlSession.execute(cql);
    }


    public static void main(String[] args) throws Exception {
        Dotenv dotenv = Dotenv.configure().directory("meta-lib-static/env/.env").load();

        //10.200.5.32
        String hosts = resolveEnv(dotenv, "ENUM_SERVER_HOSTS__STREAMING", "127.0.0.1");
        String myIp = resolveMyIp(hosts);
        if (myIp == null) {
            error("ENUM_SERVER_HOSTS__STREAMING not match! skip");
            System.exit(0);
        }

        // alikafka-pre-cn-7mz2tzhtf001-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-7mz2tzhtf001-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-7mz2tzhtf001-3-vpc.alikafka.aliyuncs.com:9092
        String kafkas = resolveEnv(dotenv, "ENUM_SERVER_KAFKAS", "127.0.0.1:9092");
        // http://zk1:8081,http://zk2:8081,http://zk3:8081
        String kafkaSrs = resolveEnv(dotenv, "ENUM_SERVER_KAFKA_SRS", "http://localhost:8081");
        // zk1:9042,zk2:9042,zk3:9042
        String cassandras = resolveEnv(dotenv, "ENUM_SERVER_CASSANDRAS__STREAMING", "");
        // data_buffer_nexus_release
        String buffer_schema = resolveEnv(dotenv, "ENUM_SERVER_BUFFER_SCHEMA", "data_buffer_dev");
        // data_pipeline_nexus_release
        String pipeline_schema = resolveEnv(dotenv, "ENUM_SERVER_PIPELINE_SCHEMA", "data_pipeline_dev");
        // event_enum_dim
        String tableId = resolveEnv(dotenv, "ENUM_SERVER_TABLE_ID", "event_enum_dim");
        // option_enum
        String optionTableId = resolveEnv(dotenv, "ENUM_SERVER_OPTION_TABLE_ID", "option_enum");

        Properties inProps = new Properties();
        inProps.put("bootstrap.servers", kafkas);
        inProps.put("auto.offset.reset", "earliest");
        inProps.put("group.id", "enum-server-java");
        inProps.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        inProps.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
        inProps.put("schema.registry.url", kafkaSrs);
        inProps.put("enable.auto.commit", "false");
        KafkaConsumer<byte[], GenericData.Array<GenericRecord>> consumer = new KafkaConsumer<byte[], GenericData.Array<GenericRecord>>(inProps);


        Properties outProps = new Properties();
        outProps.put("bootstrap.servers", kafkas);
        outProps.put("compression.type", "snappy");
        outProps.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
        outProps.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
        outProps.put("schema.registry.url", kafkaSrs);
        KafkaProducer<byte[], GenericRecord> producer = new KafkaProducer<byte[], GenericRecord>(outProps);

        CqlSessionBuilder sqlSessionBuilder = CqlSession.builder();
        for (String node : cassandras.split(",")) {
            String[] nodeParts = node.split(":");
            String host = nodeParts[0];
            int port = Integer.parseInt(nodeParts[1]);
            sqlSessionBuilder.addContactPoint(
                    new InetSocketAddress(host, port)
            );
        }
        CqlSession cqlSession = sqlSessionBuilder
                .withLocalDatacenter("datacenter1")
                .withKeyspace(CqlIdentifier.fromCql(pipeline_schema))
                .build();

        String eventEnumPkStr = "project_id,app_id,app_version,event_code";
        ClusterEnumState eventClusterState = loadEnumState(cqlSession, "event_enum_dim", null, eventEnumPkStr, "event_enum_key");
        eventClusterState.printMaxEnumKey();

        ClusterEnumState optionClusterState = loadEnumState(cqlSession, "option_enum", "cluster_name,cluster_val", "option_name,option_val", "enum_key");
        optionClusterState.printMaxEnumKey();

        // data_buffer_nexus_release_dc_sdk_push
        String inTopic = String.format("%s_dc_sdk_push", buffer_schema);
        // data_pipeline_nexus_release_event_enum_dim
        String outTopic = String.format("%s_%s", pipeline_schema, tableId);
        // data_pipeline_nexus_release_option_enum
        String outOptionTopic = String.format("%s_%s", pipeline_schema, optionTableId);

        // meta-lib-static/avsc/data_pipeline/event_enum_dim.avsc
        String schemaPath = String.format("meta-lib-static/avsc/data_pipeline/%s.avsc", tableId);
        Schema eventEnumDimSchema = new Schema.Parser().parse(new File(schemaPath));
        // meta-lib-static/avsc/data_pipeline/option_enum.avsc
        String optionSchemaPath = String.format("meta-lib-static/avsc/data_pipeline/%s.avsc", optionTableId);
        Schema optionEnumSchema = new Schema.Parser().parse(new File(optionSchemaPath));

        info(" [*] process kafka...");
        consumer.subscribe(Arrays.asList(inTopic));
        while (true) {
            for (ConsumerRecord<byte[], GenericData.Array<GenericRecord>> record : consumer.poll(Duration.ofMillis(1000))) {
                for (GenericRecord platform_record : record.value()) {
                    try {
                        for (Map<String, Object> optionJson : doOptionEnumFromCommon(optionClusterState, platform_record)) {
                            writeOptionEnum(optionJson, optionEnumSchema, producer, outOptionTopic, cqlSession);
                        }
                    } catch (Exception e) {
                        error(e.getMessage(), e);
                        e.printStackTrace();
                    }
                    // 将events可枚举属性写入outOptionTopic
                    try {
                        for (Map<String, Object> optionJson : doOptionEnumFromEvent(optionClusterState, platform_record)) {
                            writeOptionEnum(optionJson, optionEnumSchema, producer, outOptionTopic, cqlSession);
                        }
                    } catch (Exception e) {
                        error(e.getMessage(), e);
                        e.printStackTrace();
                    }
                    Map<String, Object> eventEnumCollector = new HashMap<String, Object>();
                    try {
                        doEventEnumFromCommon(platform_record, eventEnumCollector);
                        for (GenericRecord event_record : (GenericData.Array<GenericRecord>) platform_record.get("events")) {
                            Map<String, Object> eventJson = doEventEnumFromEvent(eventClusterState, event_record, eventEnumCollector);
                            if (eventJson != null)
                                writeEventEnum(eventJson, eventEnumDimSchema, producer, outTopic, cqlSession);
                        }
                    } catch (Exception e) {
                        error(e.getMessage(), e);
                        e.printStackTrace();
                    }
                }
                consumer.commitSync(Duration.ofMillis(5000));
            }
        }
    }
}

